package org.example.api.transform;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.api.bean.SensorReading;

/**
 * @author huangqihan
 * @date 2021/2/19
 */
public class TransformOnRichFunction {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SensorReading> mapStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        SingleOutputStreamOperator<Tuple2<String, String>> richMap = mapStream.map(new RichMapper());

        richMap.print();

        env.execute();
    }

    static class RichMapper extends RichMapFunction<SensorReading, Tuple2<String,String>> {

        @Override
        public Tuple2<String,String> map(SensorReading s) {
            return new Tuple2<>(s.getId(),getRuntimeContext().getTaskName());
        }

        @Override
        public void open(Configuration parameters) {
            // initial definition or create connections
            System.out.println("open");
        }

        @Override
        public void close() {
            System.out.println("close");
        }
    }
}
